package com.bfxy.rocketmq.transaction;

import com.bfxy.rocketmq.common.Contants;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.*;

/**
 * @author wangyi
 * 分布式事务消息生产者
 */
@Slf4j
public class transactionProducer {

    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, InterruptedException {
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("text_tx_producer_group_name");
        transactionMQProducer.setVipChannelEnabled(false);
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("text_tx_producer_group_name"+"-check-thread");
                return thread;
            }
        });
        //给生产者设置nameServer
        transactionMQProducer.setNamesrvAddr(Contants.NAMESRV_ADDR);
        transactionMQProducer.setExecutorService(executorService);

        //这个对象主要做两件事情 第一件事情就是异步的执行本地事务 第二件事情就是做回查
        TransactionListener transactionListener=new TransactionListenerImpl();
        transactionMQProducer.setTransactionListener(transactionListener);
        transactionMQProducer.start();
        Message message=new Message("test_tx_topic","TagA","Key",("hello rocketmq 4 tx").getBytes(RemotingHelper.DEFAULT_CHARSET));
        transactionMQProducer.sendMessageInTransaction(message,"我是回调的参数");
        Thread.sleep(Integer.MAX_VALUE);
        transactionMQProducer.shutdown();
    }

}
